package com.http.handler;

import com.alibaba.fastjson.JSON;
import com.common.code.msgpack.MsgPackDecode;
import com.common.code.msgpack.MsgPackEncode;
import com.common.constant.CommonConstant;
import com.common.message.vo.BaseMessage;
import com.tcp.handler.HttpToTcpHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static io.netty.handler.codec.http.HttpHeaders.Names.*;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;

public class HttpHandler extends ChannelInboundHandlerAdapter {

    private final static Logger logger = LoggerFactory.getLogger(HttpHandler.class);

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            FullHttpRequest request = (FullHttpRequest) msg;
            logger.info("uri {}", request.getUri());
            if (CommonConstant.HTTP_HEALTH_PATH.equalsIgnoreCase(request.getUri())) {
                // to do redis
                logger.info("get client info ");

                //TO DO response
            } else if (msg instanceof HttpContent) {
                HttpContent httpContent = (HttpContent) msg;
                ByteBuf content = httpContent.content();
                String result = content.toString(CharsetUtil.UTF_8);
                BaseMessage baseMessage = JSON.parseObject(result, BaseMessage.class);
                if (baseMessage != null && StringUtils.isNoneBlank(baseMessage.getContent())) {
                    logger.info("HTTP SERVER receive message {}", baseMessage.toString());
                    this.responseHandle(ctx, msg);
                    //TO DO HANDLE CONTENT
                    this.sendTcp(baseMessage);
                }
            }
        } finally {
//            ReferenceCountUtil.release(msg);
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        ctx.fireExceptionCaught(cause);
        logger.error("[ERROR-HttpHandler]--{}", cause);
        ctx.close();
    }

    /**
     * 发送TCP端口
     *
     * @param baseMessage
     * @throws InterruptedException
     */
    private void sendTcp(BaseMessage baseMessage) throws InterruptedException {
        Bootstrap bootstrap = null;
        NioEventLoopGroup workGroup = null;
        ChannelFuture future = null;
        try {
            bootstrap = new Bootstrap();
            workGroup = new NioEventLoopGroup(4);
            bootstrap.group(workGroup).channel(NioSocketChannel.class)
//                    .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline p = socketChannel.pipeline();
//                        p.addLast(new IdleStateHandler(0, 0, 5));
//                            p.addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, -4, 0));
                            p.addLast(new MsgPackDecode());
                            p.addLast(new MsgPackEncode());
                            p.addLast(new HttpToTcpHandler());
//                        p.addLast(new com.tcp.handler.SocketHandler());
                        }
                    });
            future = bootstrap.connect(CommonConstant.DEFAULT_TCP_HOST, CommonConstant.DEFAULT_TCP_PORT);
            ChannelFuture finalFuture = future;
            future.addListener(new ChannelFutureListener() {
                public void operationComplete(ChannelFuture futureListener) throws Exception {
                    if (futureListener.isSuccess()) {
                        logger.info("http connect tcp " + finalFuture.channel().remoteAddress());
                        logger.info("send message: {}", baseMessage.toString());
                        finalFuture.channel().writeAndFlush(baseMessage);
                        //future.channel().close();
                    }
                }
            });
//            future.channel().closeFuture().sync();
        } catch (Exception e) {
            logger.error("[ERROR--sendTcp]:{}", e);
            future.channel().closeFuture().sync();
            workGroup.shutdownGracefully().sync();
        } finally {
//            workGroup.shutdownGracefully().sync();
        }
    }

    private void responseHandle(ChannelHandlerContext ctx, Object msg) {
        FullHttpResponse response = new DefaultFullHttpResponse(
                HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer("SEND SUCCESS"
                .getBytes()));
        response.headers().set(CONTENT_TYPE, "text/plain");
        response.headers().set(CONTENT_LENGTH,
                response.content().readableBytes());
        response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        ctx.write(response);
        ChannelFuture future = ctx.writeAndFlush(
                LastHttpContent.EMPTY_LAST_CONTENT);

        if (msg instanceof FullHttpRequest) {
            HttpRequest request = (HttpRequest) msg;
            boolean keepAlive = HttpHeaders.isKeepAlive(request);
            if (!keepAlive) {
                future.addListener(ChannelFutureListener.CLOSE);
            }
        }

//        ctx.flush();
    }
}
